package atguigu.com.edu.app.dws;

import atguigu.com.edu.bean.KeywordBean;
import atguigu.com.edu.common.GmallConstant;
import atguigu.com.edu.func.KeywordUDTF;
import atguigu.com.edu.util.KeywordUtil;
import atguigu.com.edu.util.MyClickhouseUtil;
import atguigu.com.edu.util.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "    common map<string,string>,\n" +
                "    page map<string,string>, \n" +
                "    ts BIGINT,\n" +
                "    rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
                "    WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND\n" +
                ") " + MyKafkaUtil.getKafkaDDL("dwd_traffic_page_log", "dws_traffic_keyword_group"));

        //tableEnv.executeSql("select * from page_log").print();

        Table search = tableEnv.sqlQuery("select\n" +
                "    page['item'] fullword,\n" +
                "    rowtime\n" +
                "from page_log where page['last_page_id']='search' or page['item_type'] = 'keyword' and page['item'] is not null"

        );

        tableEnv.createTemporaryView("search_table",search);

       //tableEnv.executeSql("select * from search_table").print();

        Table splitTable = tableEnv.sqlQuery("SELECT keyword,rowtime\n" +
                "FROM search_table,LATERAL TABLE(ik_analyze(fullword)) t(keyword)");

        tableEnv.createTemporaryView("split_table", splitTable);

        Table resTable = tableEnv.sqlQuery("select \n" +
                "    DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "    DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss')  edt,\n" +
                "    '" + GmallConstant.KEYWORD_SEARCH + "' source,\n" +
                "    keyword, \n" +
                "    count(*) keyword_count,\n" +
                "    UNIX_TIMESTAMP()*1000 ts\n" +
                "from split_table group by keyword,TUMBLE(rowtime, INTERVAL '10' SECOND)");

        DataStream<KeywordBean> keywordDS = tableEnv.toAppendStream(resTable, KeywordBean.class);


        keywordDS.print(">>>>");

       keywordDS.addSink(MyClickhouseUtil.<KeywordBean>getSinkFunction("insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?,?)"));


        env.execute();

    }
}
